package org.example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{StringType, StructField, StructType}

import java.util.Properties

object lianxi4 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()

    // 连接MySQL数据库
    val properties = new Properties()
    properties.setProperty("user", "root")
    properties.setProperty("password", "123456")
    properties.setProperty("driver", "com.mysql.cj.jdbc.Driver") // MySQL 8.0版本需要使用com.mysql.cj.jdbc.Driver
    properties.setProperty("characterEncoding", "UTF-8") // 设置数据库连接的编码为UTF-8

    // 定义CSV文件路径
    val csvFilePath = "src/main/resources/yun2.csv" // 替换为实际的CSV文件路径

    // 定义CSV文件的schema，仅包含需要的列
    val schema = StructType(
      Array(
        StructField("学生姓名", StringType, nullable = true),    // 姓名
        StructField("学号/工号", StringType, nullable = true),  // 学号
        StructField("平时成绩", StringType, nullable = true)    // 平时成绩
      )
    )

    // 读取CSV文件，指定编码为UTF-8
    val csvDF = spark.read
      .option("header", "false") // 如果CSV文件没有表头，设置为false
      .option("encoding", "UTF-8") // 指定文件编码为UTF-8
      .schema(schema)
      .csv(csvFilePath)

    // 查看读取的CSV数据
    csvDF.show()

    // 将CSV数据追加到MySQL数据库的spark表中
    csvDF.write
      .mode("append") // 设置为追加模式
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/test?verifyServerCertificate=false&useSSL=false&characterEncoding=UTF-8")
      .option("dbtable", "spark")
      .option("user", "root")
      .option("password", "123456")
      .save()

    // 从MySQL数据库读取数据，验证写入是否成功
    val mysqlScore = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/test?verifyServerCertificate=false&useSSL=false&characterEncoding=UTF-8")
      .option("dbtable", "spark")
      .option("user", "root")
      .option("password", "123456")
      .load()

    mysqlScore.show()

    spark.stop()
  }
}